package com.sp.reactive;

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 1-5 静态创建
 * 6-7 动态创建
 *
 * @author 宫静雨
 * @version 1.0
 * @since 2023-04-16 15:17:35
 */
public class FluxTest {

    @Test
    public void test7() {
        Flux.create(fluxSink -> {
            for (int i = 0; i < 3; i++) {
                fluxSink.next(i);
            }
            fluxSink.complete();
        }).subscribe(System.out::println);
    }

    @Test
    public void test6() {
        Flux.generate(synchronousSink -> {
            synchronousSink.next("Hello");
            synchronousSink.complete();
        }).subscribe(System.out::println);
    }

    @Test
    public void test5() {
        Flux.empty();
        Flux.error(new RuntimeException());
        Flux.never();
    }

    @Test
    public void test4() throws Exception {
        Flux.interval(Duration.ofSeconds(2), Duration.ofSeconds(1))
                .subscribe(System.out::println);

        TimeUnit.SECONDS.sleep(10);
    }

    @Test
    public void test3() {
        Flux.range(1, 10)
                .subscribe(System.out::println)
                .dispose();
    }

    @Test
    public void test2() {
        Flux.fromArray(new Integer[]{1, 2, 3})
                .subscribe(System.out::println)
                .dispose();
    }

    @Test
    public void test1() {
        Flux.just(1, 1, 2, 3, 4)
                .map(x -> x * 2)
                .collect(Collectors.toList())
                .subscribe(System.out::println);
    }
}
